package com.fanxl.flink.datastream.transformation;


import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;
import java.util.Map;

/**
 * todo 水平线的使用，一直需要认真理解原理
 *
 * @author Administrator
 * @Title:
 * @Description: 数据源处理类
 * @date 2021/11/25
 */

public class WordPeriodicWatermark<T> implements AssignerWithPeriodicWatermarks<Map<String, Object>> {


    private long currentTimestamp = Long.MIN_VALUE;

    @Override
    public long extractTimestamp(Map<String, Object> word, long previousElementTimestamp) {
        long timestamp =  (Long)word.get("timeStamp");
        currentTimestamp = Math.max(timestamp, currentTimestamp);
        return timestamp;
    }

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        long maxTimeLag = 5000;
        return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - maxTimeLag);
    }


}
